package top.codedance.iotp.client.plugins.async.mqtt;

import com.alibaba.fastjson.JSONObject;
import top.codedance.iotp.client.ClientApplication;
import top.codedance.iotp.common.executer.AsyncAbstractExecuter;
import top.codedance.iotp.common.entity.CommonEntity;
import org.apache.log4j.Logger;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

public class MqttService extends Service {
    private Logger logger = ClientApplication.getLogger(MqttService.class);
    private static MqttClient mc;
    private String uri;

    public MqttService(AsyncAbstractExecuter executer, MqttServerParam param) {
        super(executer, param);
        uri = "tcp://" + param.getIp() + ":" + param.getPort();
    }

    /**
     * 创建与mqtt的联接
     */
    public synchronized void createConnect() throws Exception {
        try {
            String clientId = UUID.randomUUID().toString();
            mc = new MqttClient(uri, clientId, new MemoryPersistence());
            MqttConnectOptions mco = new MqttConnectOptions();
            mco.setCleanSession(false);
            mco.setAutomaticReconnect(true);
            mco.setUserName(param.getUsername());
            mco.setPassword(param.getPassword().toCharArray());
            mco.setConnectionTimeout(5);
            mco.setKeepAliveInterval(Integer.MAX_VALUE);
            mc.connect(mco);
            if (mc.isConnected()) {
                mc.setCallback(new MqttCallback() {
                    @Override
                    public void connectionLost(Throwable throwable) {
                        logger.debug("disconnected.");
                        try {
                            mc.reconnect();
                        }catch (Exception e){
                            logger.debug(e.getMessage());
                        }
                    }
                    @Override
                    public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                        String message = new String(mqttMessage.getPayload());
                        JSONObject jsonObject = new JSONObject();
                        jsonObject.put("topic", s);
                        jsonObject.put("message", message);
                        receipt(s, jsonObject);
                        logger.debug(s + ":" + message);
                    }
                    @Override
                    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                        logger.debug("publisher success.");
                    }
                });

                try {
                    for (String topic : param.getTopic()) {
                        mc.subscribe(topic);
                    }
                } catch (Exception exception) {
                    logger.debug(exception.getMessage());
                }
                logger.info("MQTT connect is success.");
            }
        } catch (MqttException e) {
            logger.debug(e.getMessage());
            throw new Exception("connect fail.");
        }
    }

    /**
     * 发送信息给所有监听本主题的客户端
     *
     * @param topic
     * @param body
     */
    public void receipt(String topic, JSONObject body) {
        synchronized (MqttExecuterImpl.class) {
            Set<Map.Entry<String, Set<String>>> entries = MqttExecuterImpl.tasks.entrySet();
            for (Map.Entry<String, Set<String>> entry : entries){
                if(entry.getKey().equals(topic)){
                    CommonEntity ce = genCommonEntity();
                    ce.setBody(body.toJSONString().getBytes(StandardCharsets.UTF_8));
                    Set<String> value = entry.getValue();
                    for (String cid : value){
                        executer.send(ce.build(), cid);
                        logger.debug("Send data to " + cid);
                    }
                    break;
                }
            }
        }
    }

    /**
     * 写入mqtt数据
     * @param topic
     * @param message
     */
    public void send(String topic, JSONObject message){
        try {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(message.getIntValue("qos"));
            mqttMessage.setPayload(message.getString("payload").getBytes());
            mc.publish(topic, mqttMessage);
        }catch (Exception e){
            logger.debug(e.getMessage());
        }
    }
}
